package com.navinfo.opentsp.platform.computing.analysis.service;

import com.navinfo.opentsp.platform.computing.analysis.entity.coldChain.ColdChainInfo;
import com.navinfo.opentsp.platform.computing.analysis.entity.coldChain.PointForColdChain;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 冷链<br/>
 * 从数据源获取数据service
 */
public class ColdChainLoadDataService implements Serializable {
    private static final List<String> COL_NAMES;

    static {
        COL_NAMES = Arrays.asList("terminalid", "gpsdate", "cargoboxtemperature1", "cargoboxtemperature2",
                "cargoboxtemperature3", "cargoboxtemperature4", "containerhumidity1", "containerhumidity2",
                "containerhumidity3", "containerhumidity4", "cargoboxdoorstatus1", "cargoboxdoorstatus2",
                "cargoboxdoorstatus3", "latitude", "longitude");
    }

    private final String pointDbName;
    private final String pointTableName;
    private final String saveDbName;
    private final String infoHql;

    public ColdChainLoadDataService(String pointDbName, String pointTableName, String saveDbName, String infoHql) {
        this.pointDbName = pointDbName;
        this.pointTableName = pointTableName;
        this.saveDbName = saveDbName;
        this.infoHql = infoHql;
    }

    /**
     * 获取上次处理过的中间信息
     */
    public JavaRDD<ColdChainInfo> loadInfo(SparkSession spark) {
        Dataset<Row> dataSet = spark.sql(String.format(infoHql, saveDbName));
        return dataSet.toJavaRDD().map(row -> {
            ColdChainInfo coldChainInfo = new ColdChainInfo();
            coldChainInfo.setTid(row.getAs("tid"));
            coldChainInfo.setType(row.getAs("type"));
            coldChainInfo.setStartTime(row.getAs("start_time"));
            coldChainInfo.setUpdateTime(row.getAs("update_time"));
            coldChainInfo.setLatitude(row.getAs("latitude"));
            coldChainInfo.setLongitude(row.getAs("longitude"));
            return coldChainInfo;
        });
    }

    /**
     * 读协议数据获取必要信息
     */
    public JavaRDD<PointForColdChain> loadPoint(SparkSession spark, final String day) {
        StringBuilder sb = new StringBuilder();
        sb.append(String.format("select %s from ", String.join(",", COL_NAMES)));
        if (StringUtils.isNotEmpty(this.pointDbName)) {
            sb.append(this.pointDbName).append(".");
        }
        sb.append(this.pointTableName);
        sb.append(String.format("\n where part_time = (%s)", day));
        sb.append(" and ").append(getWhereSql0200());

        return toJavaRdd(spark.sql(sb.toString()));
    }

    private JavaRDD<PointForColdChain> toJavaRdd(Dataset<Row> dataSet) {
        return dataSet.toJavaRDD().mapPartitions((FlatMapFunction<Iterator<Row>, PointForColdChain>) iterator -> new Iterator<PointForColdChain>() {
                    @Override
                    public boolean hasNext() {
                        return iterator.hasNext();
                    }

                    @Override
                    public PointForColdChain next() {
                        return getPoint(iterator.next());
                    }
                }
        );
    }

    private PointForColdChain getPoint(Row row) {
        PointForColdChain point = new PointForColdChain();
        point.setTid(row.getAs("terminalid"));
        point.setGpsTime(((long) row.getAs("gpsdate")) * 1000L);
        point.setLatitude(row.getAs("latitude"));
        point.setLongitude(row.getAs("longitude"));
        point.setCargoboxtemperature1(row.getAs("cargoboxtemperature1"));
        point.setCargoboxtemperature2(row.getAs("cargoboxtemperature2"));
        point.setCargoboxtemperature3(row.getAs("cargoboxtemperature3"));
        point.setCargoboxtemperature4(row.getAs("cargoboxtemperature4"));
        point.setContainerhumidity1(row.getAs("containerhumidity1"));
        point.setContainerhumidity2(row.getAs("containerhumidity2"));
        point.setContainerhumidity3(row.getAs("containerhumidity3"));
        point.setContainerhumidity4(row.getAs("containerhumidity4"));
        point.setCargoboxdoorstatus1(row.getAs("cargoboxdoorstatus1"));
        point.setCargoboxdoorstatus2(row.getAs("cargoboxdoorstatus2"));
        point.setCargoboxdoorstatus3(row.getAs("cargoboxdoorstatus3"));
        return point;
    }

    /**
     * 获取过滤条件
     */
    private String getWhereSql0200() {
        return "gpsdate!=0 and gpsdate is not null and terminalid!=0 and terminalid is not null ";
    }
}
